{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load_ext autoreload\n",
    "%autoreload 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    " \n",
    "import sys\n",
    "sys.path.append(\"..\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "C:\\Users\\argenisleon\\Anaconda3\\lib\\site-packages\\dask\\config.py:161: YAMLLoadWarning: calling yaml.load() without Loader=... is deprecated, as the default Loader is unsafe. Please read https://msg.pyyaml.org/load for full details.\n",
      "  data = yaml.load(f.read()) or {}\n",
      "C:\\Users\\argenisleon\\Anaconda3\\lib\\site-packages\\statsmodels\\compat\\pandas.py:49: FutureWarning: The Panel class is removed from pandas. Accessing it from the top-level namespace will also be removed in the next version\n",
      "  data_klasses = (pandas.Series, pandas.DataFrame, pandas.Panel)\n"
     ]
    }
   ],
   "source": [
    "from optimus import Optimus"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "op = Optimus(\"dask\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table style=\"border: 2px solid white;\">\n",
       "<tr>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3>Client</h3>\n",
       "<ul>\n",
       "  <li><b>Scheduler: </b>tcp://127.0.0.1:52780\n",
       "  <li><b>Dashboard: </b><a href='http://127.0.0.1:8787/status' target='_blank'>http://127.0.0.1:8787/status</a>\n",
       "</ul>\n",
       "</td>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3>Cluster</h3>\n",
       "<ul>\n",
       "  <li><b>Workers: </b>4</li>\n",
       "  <li><b>Cores: </b>8</li>\n",
       "  <li><b>Memory: </b>17.06 GB</li>\n",
       "</ul>\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "<Client: scheduler='tcp://127.0.0.1:52780' processes=4 cores=8>"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "op.client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# !pip install prefect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "import datetime\n",
    "import os\n",
    "\n",
    "import prefect\n",
    "from prefect import task\n",
    "from prefect.engine.signals import SKIP\n",
    "from prefect.tasks.shell import ShellTask"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>INCIDENT_NUMBER</th>\n",
       "      <th>OFFENSE_CODE</th>\n",
       "      <th>OFFENSE_CODE_GROUP</th>\n",
       "      <th>OFFENSE_DESCRIPTION</th>\n",
       "      <th>DISTRICT</th>\n",
       "      <th>REPORTING_AREA</th>\n",
       "      <th>SHOOTING</th>\n",
       "      <th>OCCURRED_ON_DATE</th>\n",
       "      <th>MONTH</th>\n",
       "      <th>DAY_OF_WEEK</th>\n",
       "      <th>HOUR</th>\n",
       "      <th>UCR_PART</th>\n",
       "      <th>STREET</th>\n",
       "      <th>Lat</th>\n",
       "      <th>Long</th>\n",
       "      <th>Location</th>\n",
       "      <th>year_0</th>\n",
       "      <th>year_1</th>\n",
       "      <th>year_2</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>I182070945</td>\n",
       "      <td>619</td>\n",
       "      <td>Larceny</td>\n",
       "      <td>LARCENY ALL OTHERS</td>\n",
       "      <td>D14</td>\n",
       "      <td>808</td>\n",
       "      <td>NaN</td>\n",
       "      <td>2018-09-02 13:00:00</td>\n",
       "      <td>9</td>\n",
       "      <td>Sunday</td>\n",
       "      <td>13</td>\n",
       "      <td>Part One</td>\n",
       "      <td>LINCOLN ST</td>\n",
       "      <td>42.357791</td>\n",
       "      <td>-71.139371</td>\n",
       "      <td>(42.35779134, -71.13937053)</td>\n",
       "      <td>2018</td>\n",
       "      <td>09</td>\n",
       "      <td>02 13:00:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>I182070943</td>\n",
       "      <td>1402</td>\n",
       "      <td>Vandalism</td>\n",
       "      <td>VANDALISM</td>\n",
       "      <td>C11</td>\n",
       "      <td>347</td>\n",
       "      <td>NaN</td>\n",
       "      <td>2018-08-21 00:00:00</td>\n",
       "      <td>8</td>\n",
       "      <td>Tuesday</td>\n",
       "      <td>0</td>\n",
       "      <td>Part Two</td>\n",
       "      <td>HECLA ST</td>\n",
       "      <td>42.306821</td>\n",
       "      <td>-71.060300</td>\n",
       "      <td>(42.30682138, -71.06030035)</td>\n",
       "      <td>2018</td>\n",
       "      <td>08</td>\n",
       "      <td>21 00:00:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>I182070941</td>\n",
       "      <td>3410</td>\n",
       "      <td>Towed</td>\n",
       "      <td>TOWED MOTOR VEHICLE</td>\n",
       "      <td>D4</td>\n",
       "      <td>151</td>\n",
       "      <td>NaN</td>\n",
       "      <td>2018-09-03 19:27:00</td>\n",
       "      <td>9</td>\n",
       "      <td>Monday</td>\n",
       "      <td>19</td>\n",
       "      <td>Part Three</td>\n",
       "      <td>CAZENOVE ST</td>\n",
       "      <td>42.346589</td>\n",
       "      <td>-71.072429</td>\n",
       "      <td>(42.34658879, -71.07242943)</td>\n",
       "      <td>2018</td>\n",
       "      <td>09</td>\n",
       "      <td>03 19:27:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>I182070940</td>\n",
       "      <td>3114</td>\n",
       "      <td>Investigate Property</td>\n",
       "      <td>INVESTIGATE PROPERTY</td>\n",
       "      <td>D4</td>\n",
       "      <td>272</td>\n",
       "      <td>NaN</td>\n",
       "      <td>2018-09-03 21:16:00</td>\n",
       "      <td>9</td>\n",
       "      <td>Monday</td>\n",
       "      <td>21</td>\n",
       "      <td>Part Three</td>\n",
       "      <td>NEWCOMB ST</td>\n",
       "      <td>42.334182</td>\n",
       "      <td>-71.078664</td>\n",
       "      <td>(42.33418175, -71.07866441)</td>\n",
       "      <td>2018</td>\n",
       "      <td>09</td>\n",
       "      <td>03 21:16:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>I182070938</td>\n",
       "      <td>3114</td>\n",
       "      <td>Investigate Property</td>\n",
       "      <td>INVESTIGATE PROPERTY</td>\n",
       "      <td>B3</td>\n",
       "      <td>421</td>\n",
       "      <td>NaN</td>\n",
       "      <td>2018-09-03 21:05:00</td>\n",
       "      <td>9</td>\n",
       "      <td>Monday</td>\n",
       "      <td>21</td>\n",
       "      <td>Part Three</td>\n",
       "      <td>DELHI ST</td>\n",
       "      <td>42.275365</td>\n",
       "      <td>-71.090361</td>\n",
       "      <td>(42.27536542, -71.09036101)</td>\n",
       "      <td>2018</td>\n",
       "      <td>09</td>\n",
       "      <td>03 21:05:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>319068</th>\n",
       "      <td>I050310906-00</td>\n",
       "      <td>3125</td>\n",
       "      <td>Warrant Arrests</td>\n",
       "      <td>WARRANT ARREST</td>\n",
       "      <td>D4</td>\n",
       "      <td>285</td>\n",
       "      <td>NaN</td>\n",
       "      <td>2016-06-05 17:25:00</td>\n",
       "      <td>6</td>\n",
       "      <td>Sunday</td>\n",
       "      <td>17</td>\n",
       "      <td>Part Three</td>\n",
       "      <td>COVENTRY ST</td>\n",
       "      <td>42.336951</td>\n",
       "      <td>-71.085748</td>\n",
       "      <td>(42.33695098, -71.08574813)</td>\n",
       "      <td>2016</td>\n",
       "      <td>06</td>\n",
       "      <td>05 17:25:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>319069</th>\n",
       "      <td>I030217815-08</td>\n",
       "      <td>111</td>\n",
       "      <td>Homicide</td>\n",
       "      <td>MURDER, NON-NEGLIGIENT MANSLAUGHTER</td>\n",
       "      <td>E18</td>\n",
       "      <td>520</td>\n",
       "      <td>NaN</td>\n",
       "      <td>2015-07-09 13:38:00</td>\n",
       "      <td>7</td>\n",
       "      <td>Thursday</td>\n",
       "      <td>13</td>\n",
       "      <td>Part One</td>\n",
       "      <td>RIVER ST</td>\n",
       "      <td>42.255926</td>\n",
       "      <td>-71.123172</td>\n",
       "      <td>(42.25592648, -71.12317207)</td>\n",
       "      <td>2015</td>\n",
       "      <td>07</td>\n",
       "      <td>09 13:38:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>319070</th>\n",
       "      <td>I030217815-08</td>\n",
       "      <td>3125</td>\n",
       "      <td>Warrant Arrests</td>\n",
       "      <td>WARRANT ARREST</td>\n",
       "      <td>E18</td>\n",
       "      <td>520</td>\n",
       "      <td>NaN</td>\n",
       "      <td>2015-07-09 13:38:00</td>\n",
       "      <td>7</td>\n",
       "      <td>Thursday</td>\n",
       "      <td>13</td>\n",
       "      <td>Part Three</td>\n",
       "      <td>RIVER ST</td>\n",
       "      <td>42.255926</td>\n",
       "      <td>-71.123172</td>\n",
       "      <td>(42.25592648, -71.12317207)</td>\n",
       "      <td>2015</td>\n",
       "      <td>07</td>\n",
       "      <td>09 13:38:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>319071</th>\n",
       "      <td>I010370257-00</td>\n",
       "      <td>3125</td>\n",
       "      <td>Warrant Arrests</td>\n",
       "      <td>WARRANT ARREST</td>\n",
       "      <td>E13</td>\n",
       "      <td>569</td>\n",
       "      <td>NaN</td>\n",
       "      <td>2016-05-31 19:35:00</td>\n",
       "      <td>5</td>\n",
       "      <td>Tuesday</td>\n",
       "      <td>19</td>\n",
       "      <td>Part Three</td>\n",
       "      <td>NEW WASHINGTON ST</td>\n",
       "      <td>42.302333</td>\n",
       "      <td>-71.111565</td>\n",
       "      <td>(42.30233307, -71.11156487)</td>\n",
       "      <td>2016</td>\n",
       "      <td>05</td>\n",
       "      <td>31 19:35:00</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>319072</th>\n",
       "      <td>142052550</td>\n",
       "      <td>3125</td>\n",
       "      <td>Warrant Arrests</td>\n",
       "      <td>WARRANT ARREST</td>\n",
       "      <td>D4</td>\n",
       "      <td>903</td>\n",
       "      <td>NaN</td>\n",
       "      <td>2015-06-22 00:12:00</td>\n",
       "      <td>6</td>\n",
       "      <td>Monday</td>\n",
       "      <td>0</td>\n",
       "      <td>Part Three</td>\n",
       "      <td>WASHINGTON ST</td>\n",
       "      <td>42.333839</td>\n",
       "      <td>-71.080290</td>\n",
       "      <td>(42.33383935, -71.08029038)</td>\n",
       "      <td>2015</td>\n",
       "      <td>06</td>\n",
       "      <td>22 00:12:00</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>319073 rows × 19 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "       INCIDENT_NUMBER  OFFENSE_CODE    OFFENSE_CODE_GROUP  \\\n",
       "0           I182070945           619               Larceny   \n",
       "1           I182070943          1402             Vandalism   \n",
       "2           I182070941          3410                 Towed   \n",
       "3           I182070940          3114  Investigate Property   \n",
       "4           I182070938          3114  Investigate Property   \n",
       "...                ...           ...                   ...   \n",
       "319068   I050310906-00          3125       Warrant Arrests   \n",
       "319069   I030217815-08           111              Homicide   \n",
       "319070   I030217815-08          3125       Warrant Arrests   \n",
       "319071   I010370257-00          3125       Warrant Arrests   \n",
       "319072       142052550          3125       Warrant Arrests   \n",
       "\n",
       "                        OFFENSE_DESCRIPTION DISTRICT REPORTING_AREA SHOOTING  \\\n",
       "0                        LARCENY ALL OTHERS      D14            808      NaN   \n",
       "1                                 VANDALISM      C11            347      NaN   \n",
       "2                       TOWED MOTOR VEHICLE       D4            151      NaN   \n",
       "3                      INVESTIGATE PROPERTY       D4            272      NaN   \n",
       "4                      INVESTIGATE PROPERTY       B3            421      NaN   \n",
       "...                                     ...      ...            ...      ...   \n",
       "319068                       WARRANT ARREST       D4            285      NaN   \n",
       "319069  MURDER, NON-NEGLIGIENT MANSLAUGHTER      E18            520      NaN   \n",
       "319070                       WARRANT ARREST      E18            520      NaN   \n",
       "319071                       WARRANT ARREST      E13            569      NaN   \n",
       "319072                       WARRANT ARREST       D4            903      NaN   \n",
       "\n",
       "           OCCURRED_ON_DATE  MONTH DAY_OF_WEEK  HOUR    UCR_PART  \\\n",
       "0       2018-09-02 13:00:00      9      Sunday    13    Part One   \n",
       "1       2018-08-21 00:00:00      8     Tuesday     0    Part Two   \n",
       "2       2018-09-03 19:27:00      9      Monday    19  Part Three   \n",
       "3       2018-09-03 21:16:00      9      Monday    21  Part Three   \n",
       "4       2018-09-03 21:05:00      9      Monday    21  Part Three   \n",
       "...                     ...    ...         ...   ...         ...   \n",
       "319068  2016-06-05 17:25:00      6      Sunday    17  Part Three   \n",
       "319069  2015-07-09 13:38:00      7    Thursday    13    Part One   \n",
       "319070  2015-07-09 13:38:00      7    Thursday    13  Part Three   \n",
       "319071  2016-05-31 19:35:00      5     Tuesday    19  Part Three   \n",
       "319072  2015-06-22 00:12:00      6      Monday     0  Part Three   \n",
       "\n",
       "                   STREET        Lat       Long                     Location  \\\n",
       "0              LINCOLN ST  42.357791 -71.139371  (42.35779134, -71.13937053)   \n",
       "1                HECLA ST  42.306821 -71.060300  (42.30682138, -71.06030035)   \n",
       "2             CAZENOVE ST  42.346589 -71.072429  (42.34658879, -71.07242943)   \n",
       "3              NEWCOMB ST  42.334182 -71.078664  (42.33418175, -71.07866441)   \n",
       "4                DELHI ST  42.275365 -71.090361  (42.27536542, -71.09036101)   \n",
       "...                   ...        ...        ...                          ...   \n",
       "319068        COVENTRY ST  42.336951 -71.085748  (42.33695098, -71.08574813)   \n",
       "319069           RIVER ST  42.255926 -71.123172  (42.25592648, -71.12317207)   \n",
       "319070           RIVER ST  42.255926 -71.123172  (42.25592648, -71.12317207)   \n",
       "319071  NEW WASHINGTON ST  42.302333 -71.111565  (42.30233307, -71.11156487)   \n",
       "319072      WASHINGTON ST  42.333839 -71.080290  (42.33383935, -71.08029038)   \n",
       "\n",
       "       year_0 year_1       year_2  \n",
       "0        2018     09  02 13:00:00  \n",
       "1        2018     08  21 00:00:00  \n",
       "2        2018     09  03 19:27:00  \n",
       "3        2018     09  03 21:16:00  \n",
       "4        2018     09  03 21:05:00  \n",
       "...       ...    ...          ...  \n",
       "319068   2016     06  05 17:25:00  \n",
       "319069   2015     07  09 13:38:00  \n",
       "319070   2015     07  09 13:38:00  \n",
       "319071   2016     05  31 19:35:00  \n",
       "319072   2015     06  22 00:12:00  \n",
       "\n",
       "[319073 rows x 19 columns]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "file = \"data/crime.csv\"\n",
    "df = op.load.csv(file, sep=\",\", error_bad_lines=False, header=True, null_value=\"null\", infer_schema='true', charset=\"latin1\").persist()\n",
    "df = df.cols.unnest(\"OCCURRED_ON_DATE\", separator=\"-\", splits=3, output_cols=\"year\", drop=False)\n",
    "df = df.cols.drop(\"YEAR\")\n",
    "df.compute()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "from prefect import Parameter, Flow\n",
    "\n",
    "@task\n",
    "def run_job():\n",
    "#     print(1)\n",
    "    logger = prefect.context.get(\"logger\")\n",
    "    logger.info(\"saving reference data...\")\n",
    "    file = \"data/crime.csv\"\n",
    "    df = op.load.csv(file, sep=\",\", error_bad_lines=False, header=True, null_value=\"null\", infer_schema='true', charset=\"latin1\").persist()\n",
    "    df = df.cols.unnest(\"OCCURRED_ON_DATE\", separator=\"-\", splits=3, output_cols=\"year\", drop=False)\n",
    "    df = df.cols.drop(\"YEAR\")\n",
    "    df.compute()\n",
    "    \n",
    "\n",
    "# schedules.clocks.IntervalClock(\n",
    "#     start_date=pendulum.datetime(\n",
    "#         2019, 1, 1, tz=\"America/New York\", interval=timedelta(days=1)\n",
    "#     )\n",
    "# )\n",
    "from prefect.schedules import IntervalSchedule\n",
    "from datetime import timedelta\n",
    "schedule = IntervalSchedule(interval=timedelta(minutes=2))\n",
    "\n",
    "with Flow(\"Optimus ETL\", schedule) as flow:\n",
    "    result = run_job()\n",
    "\n",
    "\n",
    "# flow.visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2020-03-23 03:35:39,488] INFO - prefect.Flow: Optimus ETL | Waiting for next scheduled run at 2020-03-23T03:36:00+00:00\n",
      "[2020-03-23 03:36:00,001] INFO - prefect.FlowRunner | Beginning Flow run for 'Optimus ETL'\n",
      "[2020-03-23 03:36:00,008] INFO - prefect.FlowRunner | Starting flow run.\n",
      "[2020-03-23 03:36:00,025] INFO - prefect.TaskRunner | Task 'run_job': Starting task run...\n",
      "[2020-03-23 03:36:00,026] INFO - prefect.Task: run_job | saving reference data...\n",
      "[2020-03-23 03:36:02,276] INFO - prefect.TaskRunner | Task 'run_job': finished task run for task with final state: 'Success'\n",
      "[2020-03-23 03:36:02,277] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded\n",
      "[2020-03-23 03:36:02,278] INFO - prefect.Flow: Optimus ETL | Waiting for next scheduled run at 2020-03-23T03:38:00+00:00\n"
     ]
    }
   ],
   "source": [
    "from prefect.engine.executors import LocalExecutor\n",
    "flow.run(executor=LocalExecutor())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Schedule https://docs.prefect.io/core/concepts/schedules.html#complex-schedules"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}